package com.atguigu.flink.sql;

import com.atguigu.flink.function.WaterSensorMapFunction;
import com.atguigu.flink.pojo.WaterSensor;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.*;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import static org.apache.flink.table.api.Expressions.*;

/**
 * Created by Smexy on 2022/12/21
 */
public class Demo9_GroupWindow
{
    public static void main(String[] args) {


        Configuration conf = new Configuration();
        conf.setInteger("rest.port", 3333);
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);

        env.getConfig().setAutoWatermarkInterval(2000);

        env.setParallelism(1);

        //为了获取eventtime
        WatermarkStrategy<WaterSensor> watermarkStrategy = WatermarkStrategy.<WaterSensor>forMonotonousTimestamps()
            .withTimestampAssigner( (e, r) -> e.getTs());

        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        SingleOutputStreamOperator<WaterSensor> ds = env
            .socketTextStream("hadoop103", 8888)
            .map(new WaterSensorMapFunction())
            .assignTimestampsAndWatermarks(watermarkStrategy)
            ;


        Table table = tableEnv.fromDataStream(ds, $("ts"),$("vc"),$("id"),$("pt").proctime(),
            $("et").rowtime());

        //给表起名字
        tableEnv.createTemporaryView("ws",table);

        /*
        https://nightlies.apache.org/flink/flink-docs-release-1.13/docs/dev/table/sql/queries/group-agg/
                SQL不支持元素个数的窗口！

                滚动：  TUMBLE(time_attr, interval)
                滑动:  HOP(time_attr, slide interval, size interval)
                会话:  SESSION(time_attr, interval)
         */
         String tumbleSql = "select  id," +
             "              TUMBLE_start(et, INTERVAL '5' SECOND)  wstart, " +
             "              TUMBLE_end(et, INTERVAL '5' SECOND)  wend, " +
             "               sum(vc) sumVc   " +
             "        from ws " +
             "        group by id, TUMBLE(et, INTERVAL '5' SECOND)  ";


        String slideSql = "select  id," +
            "              HOP_start(pt, INTERVAL '3' SECOND,INTERVAL '5' SECOND)   wstart, " +
            "             HOP_end(pt, INTERVAL '3' SECOND,INTERVAL '5' SECOND)  wend, " +
            "               sum(vc) sumVc   " +
            "        from ws " +
            "        group by id, HOP(pt, INTERVAL '3' SECOND,INTERVAL '5' SECOND)  ";


        String slideSql2 = "select  id," +
            "              HOP_start(et, INTERVAL '3' SECOND,INTERVAL '5' SECOND)   wstart, " +
            "             HOP_end(et, INTERVAL '3' SECOND,INTERVAL '5' SECOND)  wend, " +
            "               sum(vc) sumVc   " +
            "        from ws " +
            "        group by id, HOP(et, INTERVAL '3' SECOND,INTERVAL '5' SECOND)  ";



        String sessionSql = "select  id," +
            "              SESSION_start(et, INTERVAL '5' SECOND)  wstart, " +
            "              SESSION_end(et, INTERVAL '5' SECOND)  wend, " +
            "               sum(vc) sumVc   " +
            "        from ws " +
            "        group by id, SESSION(et, INTERVAL '5' SECOND)  ";


         tableEnv.executeSql(slideSql2)
                 .print();

    }
}
